package org.example.memora.task;

import jakarta.annotation.PostConstruct;
import org.springframework.data.redis.connection.stream.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.example.memora.entity.PostLike;
import org.example.memora.mapper.PostLikeMapper;
import org.example.memora.service.PostLikeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import static org.example.memora.common.constant.LikeConstant.POST_LIKE_TASK_TIME;
import static org.example.memora.common.constant.LikeConstant.POST_MAX_WRITE_SIZE;

/**
 * @author 丁利军
 */
@Component
@Slf4j
public class PostLikeRecordFlushTask {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private PostLikeMapper postLikeMapper;

    private static final String STREAM_KEY = "stream:post:like";
    private static final String GROUP_NAME = "post-like-group";
    private static final String CONSUMER_NAME = "consumer-1";

    @PostConstruct
    public void initGroup() {
        try {
            redisTemplate.opsForStream().createGroup(STREAM_KEY, ReadOffset.from("0-0"), GROUP_NAME);
            log.info("消费组 [{}] 创建成功", GROUP_NAME);
        } catch (Exception e) {
            log.warn("消费组 [{}] 已存在或创建失败: {}", GROUP_NAME, e.getMessage());
        }
    }

    @Scheduled(fixedDelay = 3000)
    public void consume() {
        //log.info("执行异步更新点赞记录");
        List<MapRecord<String, Object, Object>> messages = redisTemplate.opsForStream()
                .read(Consumer.from(GROUP_NAME, CONSUMER_NAME),
                        StreamReadOptions.empty().count(100).block(Duration.ofSeconds(3)),
                        StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()));
        if (messages == null || messages.isEmpty()) return;

        List<PostLike> insertBatch = new ArrayList<>();
        List<PostLike> deleteBatch = new ArrayList<>();

        for (MapRecord<String, Object, Object> msg : messages) {
            String userIdStr = msg.getValue().get("userId").toString();
            String postId = msg.getValue().get("postId").toString();
            String action = msg.getValue().get("action").toString();

            Long userId = Long.valueOf(userIdStr);
            PostLike like = new PostLike();
            like.setUserId(userId);
            like.setPostId(postId);
            like.setCreatedAt(Instant.now());

            try {
                if ("like".equals(action)) {
                    insertBatch.add(like);
                } else if ("unlike".equals(action)) {
                    deleteBatch.add(like);
                }
                redisTemplate.opsForStream().acknowledge(STREAM_KEY, GROUP_NAME, msg.getId());
            } catch (Exception e) {
                log.error("处理评论点赞消息异常：{}", e.getMessage(), e);
            }

            // ACK 确认处理成功
            redisTemplate.opsForStream().acknowledge(STREAM_KEY, GROUP_NAME, msg.getId());
        }

        if (!insertBatch.isEmpty()) {
            postLikeMapper.batchInsert(insertBatch);

            log.info("批量写入点赞记录 {} 条", insertBatch.size());
        }

        for (PostLike like : deleteBatch) {
            postLikeMapper.deleteLike(like.getUserId(), like.getPostId());
            log.info("删除点赞记录 userId={}, postId={}", like.getUserId(), like.getPostId());
        }
    }

//    @Autowired
//    private StringRedisTemplate redisTemplate;
//
//    @Autowired
//    private PostLikeService postLikeService; // 或直接调用 flushLikeRecords 方法所在的 Service
//
//    // 每分钟扫描一次LIKE_TASK_TIME
//    @Scheduled(fixedDelay = POST_LIKE_TASK_TIME)
//    public void flushAllLikeRecords() {
//        Set<String> keys = redisTemplate.keys("post:like:record:*");
//        if (keys == null || keys.isEmpty()) {
//            return;
//        }
//        if(keys.size()>=POST_MAX_WRITE_SIZE){
//            for (String key : keys) {
//                // 从 key 中解析 postId
//                String postId = key.substring("post:like:record:".length());
//                // 调用 flush 方法
//                postLikeService.flushLikeRecords(postId);
//                log.info("更新数据库帖子点赞记录表");
//            }
//        }
//
//    }
}
